use std::{
    collections::{hash_map::DefaultHasher, HashMap},
    hash::{Hash, Hasher},
    io,
    net::IpAddr,
    time::Duration,
};

// use libp2p::{bytes::BufMut, identity, kad, noise, swarm::SwarmEvent, tcp, yamux, PeerId};

use either::Either;
use libp2p::{
    autonat,
    gossipsub::{self, IdentTopic},
    identify,
    identity::Keypair,
    kad::{self, store::MemoryStore},
    mdns,
    multiaddr::Protocol,
    ping, relay,
    request_response::{self, OutboundRequestId, ResponseChannel},
    swarm::NetworkBehaviour,
    Multiaddr, PeerId,
};
use log::{debug, info};
use void::Void;

use crate::{config::ReqRespConfig, error::P2pError};

mod req_resp;

use req_resp::GenericCodec;
pub use req_resp::ResponseType;

/// QuantsBehaviourErr combines all possible errors generated by the subbehaviours.
pub type QuantsBehaviourErr =
    Either<Either<Either<Either<Either<io::Error, io::Error>, Void>, Void>, Void>, Void>;

/// to find peers enable to speedup establish connection between the nodes,
/// Find a peer by its peer ID
pub fn wellknown_peers() -> Vec<&'static str> {
    vec![
        // local company peer
        "12D3KooWGzkJMJbAUZAZfo3q6UTjxEXdxCFgYiyw4JPj8DzZJvws",
        "12D3KooWKP6gVLRkwnDJP1Nc4pq6tbG4gXU7rzucTt51q856rNwh",
        // local home peer
        "12D3KooWKqAWX3dd1y8cTZ8oyN5WX4WpqUbzCkNFC1bGpaWYBCZq",
        "12D3KooWSMWCW1L7zz26WSJhPNAE6tmt9C15qDuk8jAjEQX4gYhC",
        //external public peer
        "12D3KooWDB389og3n28AYneCYW1FwArf3W7ksDa4QqNtPyqYooWU",
        "12D3KooWHdUGiAq13StkL81uct7gbw2ohUk2BPktUTWCcZSixBzK", // public peer supports "/libp2p/circuit/relay/0.2.0/hop"
        "12D3KooWRraQYLZ4C4wFC4fHvNjqaa6fNxnsuAsoznSYb2eqT58r",
        "12D3KooWSjNuyZF1PYUxtKtUmvGDEz5FH6tMoiYrn6w54yYMc1tq",
        //
        // local company peer,
    ]
}

pub fn autonat_server_peers() -> Vec<&'static str> {
    vec![
        //external public peer
        "12D3KooWDB389og3n28AYneCYW1FwArf3W7ksDa4QqNtPyqYooWU",
        "12D3KooWHdUGiAq13StkL81uct7gbw2ohUk2BPktUTWCcZSixBzK", // public peer supports "/libp2p/circuit/relay/0.2.0/hop"
        "12D3KooWRraQYLZ4C4wFC4fHvNjqaa6fNxnsuAsoznSYb2eqT58r",
        "12D3KooWSjNuyZF1PYUxtKtUmvGDEz5FH6tMoiYrn6w54yYMc1tq",
    ]
}

#[derive(NetworkBehaviour)]
pub struct QuantsBehaviour {
    // `kad`, `identify`, and `ping` are used for peer discovery.
    kad: kad::Behaviour<MemoryStore>,
    identify: identify::Behaviour,
    ping: ping::Behaviour,
    // mdns: mdns::tokio::Behaviour,

    // `req_resp` is used for sending requests and responses.
    req_resp: request_response::Behaviour<GenericCodec>,

    // `pubsub` is used for broadcasting messages.
    pubsub: gossipsub::Behaviour,
    auto_nat: autonat::Behaviour,

    relay: relay::Behaviour,
}

impl QuantsBehaviour {
    pub fn new(
        local_key: Keypair,
        pubsub_topics: Vec<String>,
        req_resp_config: Option<ReqRespConfig>,
    ) -> Result<Self, P2pError> {
        let local_pubkey = local_key.public();
        let local_peer_id = local_pubkey.to_peer_id();

        // Create a Kademlia behaviour.
        let mut cfg = kad::Config::default();
        cfg.set_query_timeout(Duration::from_secs(2 * 60));
        cfg.set_caching(kad::Caching::Enabled { max_peers: 3 });

        // cfg.set_query_timeout(Duration::from_secs(5 * 60));
        let store = kad::store::MemoryStore::new(local_peer_id);
        let kad_behaviour = kad::Behaviour::with_config(local_peer_id, store, cfg);

        let id_behaviour = identify::Behaviour::new(
            identify::Config::new("/fuquants/identify/1.0.0".to_string(), local_pubkey)
                .with_interval(Duration::from_secs(5))
                .with_push_listen_addr_updates(true),
        );

        let mut autonat_behaviour = autonat::Behaviour::new(
            local_peer_id,
            autonat::Config {
                retry_interval: Duration::from_secs(2 * 6),
                refresh_interval: Duration::from_secs(1 * 6),
                boot_delay: Duration::from_secs(5),
                throttle_server_period: Duration::ZERO,
                only_global_ips: true,
                ..Default::default()
            },
        );
        for peer_id in autonat_server_peers() {
            let autonat_server_peer_id: PeerId = peer_id.parse()?;
            autonat_behaviour.add_server(autonat_server_peer_id, None);
        }

        let relay_behaviour = relay::Behaviour::new(local_peer_id, Default::default());

        Ok(Self {
            kad: kad_behaviour,
            identify: id_behaviour,
            ping: ping::Behaviour::default(),
            req_resp: Self::new_req_resp(req_resp_config),
            pubsub: Self::new_gossipsub(local_key, pubsub_topics)?,
            // mdns: Self::new_mdns(local_peer_id)?,
            auto_nat: autonat_behaviour,
            relay: relay_behaviour,
        })
    }

    /// To find peers enable to speedup establishing connection between the nodes
    pub fn discover_peers(&mut self) {
        debug!("\n\n get_closest_peers start");

        for pstr in wellknown_peers() {
            let peer_id_to_find: PeerId = pstr.parse().unwrap();
            let query_id = self.kad.get_closest_peers(peer_id_to_find);
            debug!("------{pstr} query_id={query_id}");
        }
        debug!("get_closest_peers end\n\n");

        if self.known_peers().is_empty() {
            info!("☕ Discovery process paused due to no boot node");
        } else {
            info!("☕ Starting a discovery process");
            let _ = self.kad.bootstrap();
        }
    }

    pub fn known_peers(&mut self) -> HashMap<PeerId, Vec<Multiaddr>> {
        let mut peers = HashMap::new();
        for b in self.kad.kbuckets() {
            for e in b.iter() {
                peers.insert(*e.node.key.preimage(), e.node.value.clone().into_vec());
            }
        }

        peers
    }

    pub fn send_request(&mut self, target: &PeerId, request: Vec<u8>) -> OutboundRequestId {
        self.req_resp.send_request(target, request)
    }

    pub fn send_response(&mut self, ch: ResponseChannel<ResponseType>, response: ResponseType) {
        let _ = self.req_resp.send_response(ch, response);
    }

    pub fn broadcast(&mut self, topic: String, message: Vec<u8>) -> Result<(), P2pError> {
        let topic = gossipsub::IdentTopic::new(topic);
        self.pubsub.publish(topic, message)?;

        Ok(())
    }

    pub fn add_address(&mut self, peer_id: &PeerId, addr: Multiaddr) {
        if can_add_to_dht(&addr) {
            // info!("☕ Adding address {} from {:?} to the DHT.", addr, peer_id);
            self.kad.add_address(peer_id, addr);
        } else {
            // info!(
            //     "☕ Can't add address {} from {:?} to the DHT.",
            //     addr, peer_id
            // );
        }
    }

    pub fn remove_peer(&mut self, peer_id: &PeerId) {
        debug!("☕ Removing peer {} from the DHT.", peer_id);
        self.kad.remove_peer(peer_id);
    }

    fn new_req_resp(config: Option<ReqRespConfig>) -> request_response::Behaviour<GenericCodec> {
        if let Some(config) = config {
            return req_resp::BehaviourBuilder::new()
                .with_connection_keep_alive(config.connection_keep_alive)
                .with_request_timeout(config.request_timeout)
                .with_max_request_size(config.max_request_size)
                .with_max_response_size(config.max_response_size)
                .build();
        }

        req_resp::BehaviourBuilder::default().build()
    }

    fn new_mdns(local_peer_id: PeerId) -> Result<mdns::tokio::Behaviour, P2pError> {
        let mdns = mdns::tokio::Behaviour::new(mdns::Config::default(), local_peer_id)
            .map_err(|err| P2pError::PubsubBuildError(err.to_string()))?;

        Ok(mdns)
    }

    fn new_gossipsub(
        local_key: Keypair,
        topics: Vec<String>,
    ) -> Result<gossipsub::Behaviour, P2pError> {
        let message_id_fn = |message: &gossipsub::Message| {
            let mut s = DefaultHasher::new();
            message.data.hash(&mut s);
            gossipsub::MessageId::from(s.finish().to_string())
        };

        let gossipsub_config = gossipsub::ConfigBuilder::default()
            .heartbeat_interval(Duration::from_secs(10))
            .validation_mode(gossipsub::ValidationMode::Strict)
            .message_id_fn(message_id_fn)
            .build()
            .map_err(|err| P2pError::PubsubBuildError(err.to_string()))?;

        let mut gossipsub = gossipsub::Behaviour::new(
            gossipsub::MessageAuthenticity::Signed(local_key),
            gossipsub_config,
        )
        .map_err(|err| P2pError::PubsubBuildError(err.to_string()))?;

        for t in topics {
            let topic = IdentTopic::new(t);
            gossipsub.subscribe(&topic)?;
        }

        Ok(gossipsub)
    }
}

fn can_add_to_dht(addr: &Multiaddr) -> bool {
    let ip = match addr.iter().next() {
        Some(Protocol::Ip4(ip)) => IpAddr::V4(ip),
        Some(Protocol::Ip6(ip)) => IpAddr::V6(ip),
        Some(Protocol::Dns(_)) | Some(Protocol::Dns4(_)) | Some(Protocol::Dns6(_)) => return true,
        _ => return false,
    };

    // info!(
    //     "ip: {ip} is_loopback={} is_unspecified={}",
    //     ip.is_loopback(),
    //     ip.is_unspecified()
    // );

    // !ip.is_loopback() && !ip.is_unspecified()
    !ip.is_unspecified()
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_can_add_to_dht() {
        let ip4_loopback: Multiaddr = "/ip4/127.0.0.1/tcp/8000".parse().unwrap();
        let ip6_loopback: Multiaddr = "/ip6/::1/tcp/8000".parse().unwrap();
        let ip4_unspecified: Multiaddr = "/ip4/0.0.0.0/tcp/0".parse().unwrap();
        let ip6_unspecified: Multiaddr = "/ip6/::/tcp/0".parse().unwrap();
        let ip4: Multiaddr = "/ip4/192.168.0.10/tcp/8000".parse().unwrap();
        let ip6: Multiaddr = "/ip6/fe80::1/tcp/8000".parse().unwrap();
        let domain_name: Multiaddr = "/dns4/example.com/tcp/8000".parse().unwrap();

        assert!(!can_add_to_dht(&ip4_loopback));
        assert!(!can_add_to_dht(&ip6_loopback));
        assert!(!can_add_to_dht(&ip4_unspecified));
        assert!(!can_add_to_dht(&ip6_unspecified));
        assert!(can_add_to_dht(&ip4));
        assert!(can_add_to_dht(&ip6));
        assert!(can_add_to_dht(&domain_name));
    }
}
